package com.atguigu.api4

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}

/**
 * @description: tableApi操作
 * @time: 2020/7/22 17:22
 * @author: baojinlong
 **/
object TableApiTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 默认就是老版本流查询环境
    // val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // 设置并行度方便测试
    env.setParallelism(1)

    // 1.1创建老版本流查询环境
    // 环境参数设置
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useOldPlanner()
      .inStreamingMode()
      .build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)


    // 1.2创建老版本批式查询环境
    val batchEnv: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    val batchTableEnv: BatchTableEnvironment = BatchTableEnvironment.create(batchEnv)

    // 1.3创建Blink版本的流查询环境
    val bsSettings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, bsSettings)

    // 1.4创建blink版本批式查询环境
    val bbSettings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inBatchMode()
      .build()
    // 下面如果能执行需要使用-target:jvm1.8
    // val environment: TableEnvironment = TableEnvironment.create(bbSettings)


    // 从外部系统读取数据,在环境中注册表.连接到外部文件系统
    val inputPath: String = "E:/qj_codes/big-data/FlinkTutorial/src/main/resources/sensor.data"

    tableEnv.connect(new FileSystem().path(inputPath))
      //.withFormat(new OldCsv())
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING)
        .field("timestamp", DataTypes.BIGINT)
        .field("temperature", DataTypes.DOUBLE)
      ) // 定义表结构
      .createTemporaryTable("inputTable") // 注册临时表


    // 转换成流打印输出
    val sensorTable: Table = tableEnv.from("inputTable")
    sensorTable.toAppendStream[(String, Long, Double)].print("inputTable-xxx")


    // 执行
    env.execute("table env job")
  }

}
